#include "systhreadpool.h"
#include "server_global.h"

SysThreadPool::SysThreadPool()
{

}

void SysThreadPool::delThread(CnsComponentPtr com, DbQueryBaseThread* thread)
{
    DBQUERYBASETHREADPTRSET threads = com->property("threads").value<DBQUERYBASETHREADPTRSET>();
    threads.remove(thread);
    thread->release();
}

bool SysThreadPool::exhaustTasks(CnsComponentPtr com)
{
    DBQUERYBASETHREADPTRSET threads = com->property("threads").value<DBQUERYBASETHREADPTRSET>();
    int poolSize = com->property("poolSize").toInt();
    QSVHASHLIST svl = com->property("threadTasks").value<QSVHASHLIST>();
    return __exhaustTasks(com, svl, poolSize, threads);
}

bool SysThreadPool::__exhaustTasks(CnsComponentPtr com, QSVHASHLIST &svl, int poolSize, DBQUERYBASETHREADPTRSET& tl)
{
    QSVHASHLIST::iterator it = svl.begin();
    int size = tl.size();
    for(int i = size; i < poolSize && it != svl.end(); i ++)
    {
        DbQueryBaseThread* pt = newQueryThread(com, *it);
        tl.insert(pt);
        pt->retain();
        it = svl.erase(it);
    }
    if(size != tl.size()){
        com->setProperty(QVariant::fromValue(tl), "threads");
        com->setProperty(QVariant::fromValue(svl), "threadTasks");
    }
    return !svl.size();
}

void SysThreadPool::f_execTask(CnsComponentPtr com, QVariantHash &sv)
{
    DBQUERYBASETHREADPTRSET threads = com->property("threads").value<DBQUERYBASETHREADPTRSET>();
    int poolSize = com->property("poolSize").toInt();
    QSVHASHLIST svl = com->property("threadTasks").value<QSVHASHLIST>();
    bool exhausted = __exhaustTasks(com, svl, poolSize, threads);
    if(exhausted && threads.size() < poolSize)
    {
        DbQueryBaseThread* pt = newQueryThread(com, sv);
        threads.insert(pt);
        com->setProperty(QVariant::fromValue(pt), "threads");

    } else{
        QSVHASHLIST svl = com->property("threadTasks").value<QSVHASHLIST>();
        svl.push_back(sv);
        com->setProperty(QVariant::fromValue(svl), "threadTasks");
    }
}

DbQueryBaseThread* SysThreadPool::newQueryThread(CnsComponentPtr com, QVariantHash& sv)
{
    static uint64_t threads = 1;

    DbQueryThreadPtr pt = FRAMEWORK->newComponent(QString("thread%1").arg(threads++),
      sv["threadPath"].toString(), sv["replacePathList"].value<REPLACEPATHLIST>()).staticCast<DbQueryThread>();
    pt->execInNs("initDbQueryThread", "",
               Q_ARG(QVariantHash&, sv), Q_ARG(QScopedPointer<QThread>&, pt->getWorkThread()));
    connect(pt.data(), &DbQueryBaseThread::dbQueryThreadFinished, this, [com, this](CnsComponentPtr tcom){
        DbQueryBaseThread* thread = static_cast<DbQueryBaseThread*>(tcom.data());
        delThread(com, thread);
        exhaustTasks(com);
        if(thread->isForceStop())return;

        tcom->exec("writeDbQueryResult", Q_ARG(const QVariantHash&, thread->getProps()));
        int ret = FRAMEWORK->removeRefObj(thread->getObjFullPath(), true);
        assert(ret);
    });

    return pt.data();
}
